基于FastApi框架测试平台(13)

您所在的位置：网站首页 › fastapi 分页功能开发 › 基于FastApi框架测试平台(13)

基于FastApi框架测试平台(13)

2024-04-06 16:04| 来源: 网络整理| 查看: 265

前言

这里对上一节遗留的问题解决一下

重复写`with SessionLocal() as session比较繁琐实现分页查询问题1

针对问题1，解决方案很简单，这里使用FastAPI中的depends。

简单介绍一下depends：

在 FastAPI 中，depends 是依赖注入系统的一部分，它可以用作声明和解析函数参数的一种方式。通过使用 depends，你可以方便地将对象、函数或类作为依赖项注入到其他函数中。

使用depends大概有以下几个步骤：

定义一个依赖项函数在路由处理函数中声明依赖，通过在路由处理函数的参数列表中使用 depends 关键字，声明需要的依赖项在需要使用依赖项的地方，使用声明的参数名称进行访问

好了，使用方法了解了，我们就按照这步骤解决问题1

定义依赖项函数

def get_session(): with SessionLocal() as session: return session

在路由处理函数中声明依赖

@router.post("/create") def role_create(data: role_schema.Role, session=Depends(get_session)): data, err = RoleService.create(session, data.name, data.comment, data.ability_ids) if err: return resp_error(USER_ERROR_A0500, msg=err) return resp_success()

使用声明的参数名访问

class RoleServiceobject: @classmethod def create(cls, session, name: str, comment: str, ability_ids: list = None): try: role = session.query(Role).filter(Role.name == name, Role.status != 0).first() ......

这样就解决了。

问题2

分页问题也好解决，对应到sql查询就是offset、limit

封装一个分页获取结果的方法：

class ServiceBase(object): @staticmethod def pagination(page: int, size: int, session, sql: str, scalars=True): data = session.execute(sql) total = data.raw.rowcount if total == 0: return [], 0 sql = sql.offset((page - 1) * size).limit(size) data = session.execute(sql) if scalars: return data.scalars().all(), total return data.all(), total

这段代码很好理解，重点提一下.scalars()和.all()的区别：

.scalars() 和 .all() 是 SQLAlchemy 中用于从查询结果集中提取数据的方法。

.scalars() 方法返回的是一个标量迭代器（scalar iterator），它会将查询的结果转换为一个个标量值，并以迭代器的形式返回。每个标量值表示查询结果中的一行数据，由于仅返回标量值，因此可以在结果集较大时提供更好的性能和内存使用效率。

.all() 方法返回的是一个列表（list），其中包含了查询结果集的所有行数据。每行数据通常是作为一个元组（tuple）或字典（dict）的形式返回，其中包含每个字段的值。这意味着返回的是完整的查询结果集。

.scalars().all() 结合使用了两个方法，首先使用 .scalars() 方法将结果集转换为标量迭代器，然后使用 .all() 方法将标量迭代器转换为列表，最终返回一个包含所有行数据的列表。

对于 .scalars().all() 的使用，常见于只需要获取查询结果中某个特定字段值的场景，比如只需要获取所有用户名或者所有订单数量等。

而对于 .all() 的使用，适用于需要获取查询结果中所有字段值的场景，可以获得完整的数据集合。

分页查询核心功能实现了，其他就简单了，

修改路由处理函数

@router.get("/list") def role_list(page: int = None, size: int = None, session=Depends(get_session)): page = page if page else 1 size = size if size else 9999 data, err = RoleService.query_all(session, page, size) ...

这里没有设置为必传，如果不传，我们给一个默认值

修改查询函数

def query_all(cls, session, page: int, size: int): conditions = [Role.status == 1] sql = cls._query().where(*conditions) try: data, total = cls.pagination(page, size, session, sql, False) ...

这样查询角色列表，就实现了分页查询。我们将分页查询单独封装，以后其他列表查询就可以直接调用。

后续

到这里，角色管理的服务端功能应该比较完善了，下面需要实现前端页面，并进行前后端联调。

【本文地址】

基于FastApi框架测试平台(13)

基于FastApi框架测试平台(13)

今日新闻

推荐新闻